package com.winit.schedule.task.tracker.utils;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.protocol.Protocol;
import com.winit.schedule.core.protocol.command.CompleteTaskRequest;
import com.winit.schedule.core.protocol.command.HeartbeatRequest;
import com.winit.schedule.core.protocol.command.HeartbeatResponse;
import com.winit.schedule.core.protocol.command.QuartzRequest;
import com.winit.schedule.core.repository.entity.QuartzJobEntity;
import com.winit.schedule.core.repository.entity.QuartzSchedulerEntity;
import com.winit.schedule.core.utils.StringUtils;
import com.winit.schedule.remoting.InvokeCallback;
import com.winit.schedule.remoting.netty.ResponseFuture;
import com.winit.schedule.remoting.protocol.RemotingCommand;
import com.winit.schedule.task.tracker.TaskTracker;
import com.winit.schedule.task.tracker.quartz.event.QuartzManagerEvent;
import com.winit.schedule.task.tracker.remoting.RemotingClientService;

/**
 * 远程调用工具类
 * 
 * @author jianke.zhang 2015-3-17 上午10:26:46
 * @since 1.0
 */
public abstract class RemoteUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(RemoteUtil.class);

    private RemoteUtil(){

    }

    /**
     * 发送心跳
     * 
     * @param remotingClient
     * @param addr
     */
    @SuppressWarnings("unchecked")
    public static boolean heartbeat(HeartbeatRequest heartBeatRequest, RemotingClientService remotingClientService,
                                    TaskTracker taskTracker) {

        if (!StringUtils.isBlank(remotingClientService.getJobTrackerAddress())) {
            RemotingCommand request = RemotingCommand.createRequestCommand(Protocol.RequestCode.HEART_BEAT.code(),
                heartBeatRequest);
            RemotingCommand response = null;
            try {
                response = remotingClientService.getNettyClient()
                    .invokeSync(remotingClientService.getJobTrackerAddress(),
                        request,
                        remotingClientService.getInvokeTimeoutMillis());
            } catch (Exception e) {
                LOGGER.error("Heartbeat failure.", e);
                return false;
            }

            if (response != null && Protocol.ResponseCode.SUCCESS == Protocol.ResponseCode.valueOf(response.getCode())) {
                // if (LOGGER.isDebugEnabled()) {
                // LOGGER.debug("heart beat success! ");
                // }

                Object body = response.getBody();
                if (null != body && body instanceof HeartbeatResponse) {
                    HeartbeatResponse pushTask = (HeartbeatResponse) body;

                    List<QuartzJobEntity> commands = pushTask.getJobCmds();
                    if (null != commands) {
                        for (QuartzJobEntity command : commands) {
                            if (null != command.getCommand()) {
                                taskTracker.getDispatcher()
                                    .getEventHandler()
                                    .handle(new QuartzManagerEvent(command.getCommand(), command));
                            }
                        }
                    }

                    List<QuartzSchedulerEntity> schedulerCmds = pushTask.getSchedulerCmds();
                    if (null != schedulerCmds) {
                        for (QuartzSchedulerEntity command : schedulerCmds) {
                            if (null != command.getCommand()) {
                                taskTracker.getDispatcher()
                                    .getEventHandler()
                                    .handle(new QuartzManagerEvent(command.getCommand(), command));
                            }
                        }
                    }

                    // 先启动任务
                    taskTracker.reAllocationTask(pushTask.getTasks());

                    // 停止任务
                    taskTracker.stopTask(pushTask.getStopTaskIds());
                }
                return true;
            } else {
                LOGGER.error("heartbeat failed! ");
            }
        } else {
            if (LOGGER.isWarnEnabled()) {
                LOGGER.warn("----- Not from zookeeper get to jobtracker address");
            }
        }

        return false;
    }

    /**
     * 发送任务完成命令（ 任务正常完成、异常终止、任务人工终止）
     * 
     * @param taskCompleteRequest
     * @param remotingClientService
     * @return
     */
    public static boolean sendTaskCompleteCommand(CompleteTaskRequest taskCompleteRequest,
                                                  RemotingClientService remotingClientService) {
        if(null==remotingClientService){
            return true;
        }
        if (!StringUtils.isBlank(remotingClientService.getJobTrackerAddress())) {
            RemotingCommand request = RemotingCommand.createRequestCommand(Protocol.RequestCode.COMPLETE_TASK.code(),
                taskCompleteRequest);
            RemotingCommand response = null;
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Send complete task requeset, task : " + taskCompleteRequest.getTask());
            }
            try {
                response = remotingClientService.getNettyClient()
                    .invokeSync(remotingClientService.getJobTrackerAddress(),
                        request,
                        remotingClientService.getInvokeTimeoutMillis());
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
                return false;
            }

            if (response != null && Protocol.ResponseCode.SUCCESS == Protocol.ResponseCode.valueOf(response.getCode())) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Complete task requeset success! ");
                }
                return true;
            }
        } else {
            if (LOGGER.isWarnEnabled()) {
                LOGGER.warn("----- Not from zookeeper get to jobtracker address");
            }
        }
        return false;

    }

    /**
     * 发送Quartz信息（日志）
     * 
     * @param taskCompleteRequest
     * @param remotingClientService
     * @return
     */
    public static void sendQuartzCommand(QuartzRequest quartzRequest, RemotingClientService remotingClientService) {
        RemotingCommand request = RemotingCommand.createRequestCommand(Protocol.RequestCode.QUARTZ_REQ.code(),
            quartzRequest);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Send quartz requeset.");
        }
        try {
            remotingClientService.getNettyClient().invokeAsync(remotingClientService.getJobTrackerAddress(),
                request,
                remotingClientService.getInvokeTimeoutMillis(),
                new InvokeCallback() {

                    @Override
                    public void operationComplete(ResponseFuture responseFuture) {
                        RemotingCommand response = responseFuture.getResponseCommand();
                        if (response != null
                            && Protocol.ResponseCode.SUCCESS == Protocol.ResponseCode.valueOf(response.getCode())) {
                            if (LOGGER.isDebugEnabled()) {
                                LOGGER.debug("Complete quartz success! ");
                            }
                        }
                    }
                });
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }
}
